package com.yujiahao.bigdata.sql

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object SparkSQL01_Hive_Request_Finally {
  def main(args: Array[String]): Unit = {

    //这个要使用hadoop中的用户名，不然会报权限不够错误：user_visit_action
    System.setProperty("HADOOP_USER_NAME", "yujiahao")
    //创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .enableHiveSupport()
      .master("local[*]")
      .appName("sql")
      .getOrCreate()
    spark.sql(" use atguigu0819")

    //TODO 1、将数据补充完整，关联三张表，连接在一起。
    spark.sql(
      """
        |select
        |a.*,
        |p.product_name,
        |c.area,
        |c.city_name
        |from user_visit_action a
        |join product_info p on a.click_product_id = p.product_id
        |join city_info c on a.city_id = c.city_id
        |where a.click_product_id != -1
        |""".stripMargin).createOrReplaceTempView("t1")
    //TODO 2、将区域和商品作为整体，统计点击数量。 (( 区域, 商品 ), sum)
    //TODO 3、将统计结果进行结构的转换。(( 区域, 商品 ), sum) => (区域, ( 商品, sum))

    //在这里使用我们自己定义的聚合函数并将城市传进去
    spark.udf.register("myCnt", functions.udaf(new MyUDF()))
    spark.sql(
      """
        |select
        |t1.area,
        |t1.product_name,
        |count(*) as cliCnt,
        |myCnt(t1.city_name)
        |from t1
        |group by t1.area ,t1.product_name
        |""".stripMargin).createOrReplaceTempView("t2")
    //TODO 4、将转换后的数据根据区域进行分组。(区域, Iter[( 商品, sum),( 商品1, sum1),( 商品2, sum2)])
    spark.sql(
      """
        |select
        |*,
        |rank() over(partition by t2.area order by t2.cliCnt desc) as rank
        |from t2
        |""".stripMargin).createOrReplaceTempView("t3")
    //TODO 5、将分组后的数据按照点击数进行排序，取前三名

    spark.sql(
      """
        |select
        |*
        |from t3
        |where rank <= 3
        |""".stripMargin).show()

    //TODO 2、关闭资源
    spark.stop()
  }

  //TODO 实现城市的话就要自己定义一个UDAF函数
  case class MyBuff(var total: Long, cityMap: mutable.Map[String, Long])

  class MyUDF extends Aggregator[String, MyBuff, String] {
    override def zero: MyBuff = {
      MyBuff(0L, mutable.Map())
    }

    override def reduce(buff: MyBuff, city: String): MyBuff = {
      //缓冲区有一个记录总和的，要累加一，还有一个城市和城市数量的集合，进来一个城市累加一个总和，累加一个城市数量
      buff.total += 1L
      val oldCnt = buff.cityMap.getOrElse(city, 0L)
      buff.cityMap.update(city, oldCnt + 1L)
      buff

    }

    override def merge(b1: MyBuff, b2: MyBuff): MyBuff = {
      //对每个缓冲区之间进行处理，总和相加
      b1.total += b2.total
      //两个集合合并
      b2.cityMap.foreach {
        case (city, cnt) => {
          //取出b1里面的数量
          val oldCnt: Long = b1.cityMap.getOrElse(city, 0L)
          //将结果更新到b1中去
          b1.cityMap.update(city, cnt + oldCnt)
        }
      }
      b1
    }

    //这里进行最终计算提交
    override def finish(reduction: MyBuff): String = {
      //先将里面的数据拿出来
      val total: Long = reduction.total
      val cityMap: mutable.Map[String, Long] = reduction.cityMap
      //因为要拿前三名所以要排序，因为Map无序所以要转成有序的list
      val sortCitys: List[(String, Long)] = cityMap.toList.sortBy(_._2)(Ordering.Long.reverse)
      val top2: List[(String, Long)] = sortCitys.take(2)

      //到这里因为最终结果是一个字符串拼接以逗号结尾的，因此我们将结果放在list集合中去
      var list = ListBuffer[String]()
      var rest = 100L
      top2.foreach {
        //这里对缓冲的商品总数，以及各地区占的商品总数进行处理
        case (city, cnt) => {
          val r = cnt * 100 / total
          rest -= r
          list.append(s"${city} ${r}%")
        }
      }
      //这里对大于2个地区的商品数进行处理,因为顺序问题要放在前面两个后面
      if (sortCitys.length > 2) {
        list.append(s"其他${rest}%")
      }
      //既然已经添加到了list那么我们将它便利出来用逗号切割就行
      list.mkString(", ")
    }

    override def bufferEncoder: Encoder[MyBuff] = Encoders.product

    override def outputEncoder: Encoder[String] = Encoders.STRING
  }
}
